package cn.cuilan.common.pipeline;

import cn.cuilan.common.client.ProtocolMessageClient;
import cn.cuilan.common.entity.ProtocolMessage;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 重连监听器
 *
 * @author zhang.yan
 */
public class ReconnectOnCloseListener<T extends ProtocolMessage> implements ChannelFutureListener {

    private final Logger logger = LoggerFactory.getLogger(ReconnectOnCloseListener.class);

    /**
     * 私有协议客户端
     */
    private final ProtocolMessageClient<T> client;

    /**
     * 重新连接间隔
     */
    private final int reconnectInterval;

    /**
     * 断连请求
     */
    private final AtomicBoolean disconnectRequested = new AtomicBoolean(false);

    private final ScheduledExecutorService executorService;

    public ReconnectOnCloseListener(ProtocolMessageClient<T> client,
                                    int reconnectInterval,
                                    ScheduledExecutorService executorService) {
        this.client = client;
        this.reconnectInterval = reconnectInterval;
        this.executorService = executorService;
    }

    /**
     * 请求重新连接
     */
    public void requestReconnect() {
        disconnectRequested.set(false);
    }

    /**
     * 请求断开连接
     */
    public void requestDisconnect() {
        disconnectRequested.set(true);
    }

    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        final Channel channel = future.channel();
        logger.debug("Client connection was closed to {}", channel.remoteAddress());
        channel.disconnect();
        scheduleReconnect();
    }

    public void scheduleReconnect() {
        if (!disconnectRequested.get()) {
            logger.trace("Failed to connect. Will try again in {} millis", reconnectInterval);
            executorService.schedule(
                    () -> client.connectAsync(),
                    reconnectInterval, TimeUnit.MILLISECONDS);
        }
    }
}
